package com.etlmaster.rowprocessor.steps;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.ResultSetExtractor;

import com.etlmaster.executor.trans.core.Line;
import com.etlmaster.executor.trans.core.LineMeta;
import com.etlmaster.jdbc.JdbcTemplate;
import com.etlmaster.rowprocessor.core.AbstractReader;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class JdbcReader  extends AbstractReader{

	private String dbName;
	private String sqlExpression;
	private String tableName;
	public JdbcReader(String id, String jobId, String name, String code,
			Map<String, Object> paramMap, Map<String, Object> var) {
		super(id, jobId, name, code, paramMap, var);
		// TODO Auto-generated constructor stub
	}

	@Override
	public void write() throws Exception {
		JdbcTemplate jdbc=new JdbcTemplate(dbName);
		{
			jdbc.query(sqlExpression,
					new ResultSetExtractor<List<Map<String, Object>>>() {
				@Override
				public List<Map<String, Object>> extractData(ResultSet rs)
						throws SQLException, DataAccessException {
					ResultSetMetaData rm = rs.getMetaData();
					int colCount = rm.getColumnCount();
					LineMeta meta=new LineMeta();
					while (rs.next()) {
						Line line = new Line(meta);
						for (int i = 1; i <= colCount; i++) {
							//updateMeta(tableName, rm.getColumnLabel(i),meta);
							line.addField(rs.getString(i),tableName, rm.getColumnLabel(i)
									);
						}
						sendData(line);
					}
					return null;
				}
			});
		}
		
	}

	@Override
	protected void init() {
		// TODO Auto-generated method stub
		
	}

	@Override
	protected void stop() {
		// TODO Auto-generated method stub
		
	}

}
